package com.github.ghsea.cc.bolt;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.github.ghsea.cc.consts.FiledName;
import com.github.ghsea.cc.module.NginxAccessLog;
import com.github.ghsea.cc.util.AccessLogParser;

/**
 * 切分nginx access log
 * 
 * @author GuHai 2017-7-27下午10:38:21
 */
public class NginxLogSplitBolt extends BaseBasicBolt {

	private Logger log = LoggerFactory.getLogger(NginxLogSplitBolt.class);

	private static final long serialVersionUID = 8673690343773171373L;

	public void execute(Tuple input, BasicOutputCollector collector) {
		String accessLog = input.getString(0);
		log.info("NginxLogSplitBolt received:" + accessLog);
		NginxAccessLog logVo = AccessLogParser.parse(accessLog);
		collector.emit(new Values(logVo.getRequestUrl(), logVo.getRequestHost(), logVo.getAccessIp(), logVo
				.getAccessTime()));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields(FiledName.REQUEST_URL, FiledName.REQUEST_HOST, FiledName.ACCESS_IP,
				FiledName.ACCESS_TIME));
	}

}
